/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#include "SBJ/Arrow/Channel.h"
#include "SBJ/Arrow/Dispatcher.h"
#include "SBJ/Arrow/NULLDispatcher.h"

namespace SBJ
{

namespace Arrow
{

Channel::Channel( const IExceptionObserver::Ptr& excObserver )
: m_pubMgr( excObserver )
, m_started( false )
, m_useDispatcher( true )
{
	init( dispatch_none, dispatch_immediately, 1, 0, 0, excObserver );
}

Channel::Channel(	int numMsgHolderThreads,
					int numDispatchingThreads,
					size_t priorityCount,
					EnqueuePolicy* enqueuePolicy,
					DequeuePolicy* dequeuePolicy,
					const IExceptionObserver::Ptr& excObserver)
: m_pubMgr( excObserver )
, m_started( false )
, m_useDispatcher( numDispatchingThreads >= 0 )
{
	init( numMsgHolderThreads, numDispatchingThreads, priorityCount, enqueuePolicy, dequeuePolicy, excObserver );
}

void
Channel::init(	int numMsgHolderThreads,
				int numDispatchingThreads,
				size_t priorityCount,
				EnqueuePolicy* enqueuePolicy,
				DequeuePolicy* dequeuePolicy,
				const IExceptionObserver::Ptr& excObserver )
{
	 m_queue.reset( new Queue( priorityCount, excObserver, enqueuePolicy, dequeuePolicy ) );

	if ( numMsgHolderThreads >= 0 )
	{
		m_msgHolder.reset( new MessageHolder( numMsgHolderThreads, excObserver ) );
		m_msgHolder->addRouter( m_queue );
		m_entry = m_msgHolder;		
	}
	else
	{
		m_entry = m_queue;	
	}

	if ( numDispatchingThreads >= 0 )
	{
		m_dispatcher.reset( new Dispatcher( numDispatchingThreads ) );
		m_queue->addRouter( m_dispatcher );
	}
	else
	{
		m_dispatcher.reset( new NULLDispatcher );
	}
}

Channel::~Channel( void )
{
	stop();
}

bool
Channel::start( void )
{
	if ( m_started == false )
	{
		m_started = true;
		m_dispatcher->resetStats();
		m_started &= m_queue->start();
		if ( m_msgHolder ) m_started &= m_msgHolder->start();
		Feeds::iterator iter( m_feeds.begin() );
		while ( iter != m_feeds.end() )
		{
			m_started &= (*iter)->start();
			++iter;
		}
	}
	return ( m_started == true );
}

bool
Channel::stop( bool join )
{
	if ( m_started == true )
	{
		Feeds::iterator iter( m_feeds.begin() );
		while ( iter != m_feeds.end() )
		{
			(*iter)->stop( join );
			++iter;
		}
		if ( m_msgHolder ) m_msgHolder->stop( join );
		m_queue->stop( join );
	}
	return ( m_started == false );
}

void
Channel::addFeed( const IPublisher::Ptr& feed, boost::tribool start  )
{
	m_feeds.push_back( feed );
	feed->addRouter( m_entry );
	if ( boost::indeterminate( start ) && m_started == true )
	{
		feed->start();
	}
	else if ( start )
	{
		feed->start();
	}
}

void
Channel::onMessage( Message& msg )
{
	m_entry->onMessage( msg );
}

Topic
Channel::getTopic( const std::string& pubKey, const std::string& topicKey ) const
{
	return m_pubMgr.getTopic( pubKey, topicKey );
}

Subscription
Channel::makeSubscription( void* subscriber, const std::string& pubKey, const std::string& topicKey, const std::string& filter )
{
	return Subscription( *this, subscriber, m_pubMgr.makeTopic( pubKey, topicKey ), filter );
}

bool
Channel::subscribe(const Subscription& subscription )
{
	if ( m_dispatcher->subscribe( subscription ) )
	{
		m_queue->addInterest( subscription );
		if ( m_msgHolder ) m_msgHolder->addInterest( subscription );
		
		Feeds::iterator iter( m_feeds.begin() );
		while ( iter != m_feeds.end() )
		{
			(*iter)->addInterest( subscription );
			++iter;
		}
		return true;
	}
	return false;
}

bool
Channel::unsubscribe( const Subscription& subscription )
{
	if ( m_dispatcher->unsubscribe( subscription ) )
	{
		m_queue->removeInterest( subscription );
		if ( m_msgHolder ) m_msgHolder->removeInterest( subscription );
		
		Feeds::iterator iter( m_feeds.begin() );
		while ( iter != m_feeds.end() )
		{
			(*iter)->removeInterest( subscription );
			++iter;
		}
		return true;
	}
	return false;
}

void
Channel::printStats( std::ostream& out ) const
{
	Queue::Stats qstats = getQueueStats();
    out << "Enqueue Stats" << std::endl;
    out << " *Discovered:   " << qstats.enqueue.discoveries << std::endl;
    out << "  Accumed:      " << qstats.enqueue.consumptions << std::endl;
    out << "  Reused:       " << qstats.enqueue.reused << std::endl;
    out << " *Collided:     " << qstats.enqueue.collisions << std::endl;
    out << " *Variants:     " << qstats.enqueue.variants << std::endl;
    out << " *Independents: " << qstats.enqueue.independents << std::endl;
    out << "  Allocations:  " << qstats.enqueue.allocations << std::endl;
    out << "  Dropped:      " << qstats.enqueue.dropped << std::endl;
    out << "  Invalid:      " << qstats.enqueue.invalid << std::endl;
    out << "  Available:    " << qstats.enqueue.available << std::endl;
    out << "  Seconds:      " << qstats.enqueue.time.seconds << std::endl;
    out << "  Total:        " << qstats.enqueue.time.total << std::endl; 
    out << "  Hits:         " << qstats.enqueue.hits << std::endl;
    out << "  Rate:         " << qstats.enqueue.time.rate << std::endl;
    out << std::endl;
    out << "Dequeue Stats" << std::endl;
    out << "  Sleeps:       " << qstats.dequeue.sleeps << std::endl; 
    out << "  Seconds:      " << qstats.dequeue.time.seconds << std::endl; 
    out << "  Total:        " << qstats.dequeue.time.total << std::endl; 
    out << "  Rate:         " << qstats.dequeue.time.rate << std::endl;
    out << std::endl;
    if ( isDispatching() == true )
    {
	IDispatcher::Stats dstats = getDispatchStats();
    out << "Dispatch Stats" << std::endl;
    out << "  Subscriptions:" << dstats.subscriptions << std::endl;
    out << "  UnSubscribe:  " << dstats.unsubscriptions << std::endl;
    out << "  Executions:   " << dstats.executions << std::endl;
    out << "  Begin Visits: " << dstats.visitBeginnings << std::endl;
    out << "  Visitations:  " << dstats.visitations << std::endl;
    out << "  Cancelations: " << dstats.cancelations << std::endl;
    out << "  Dropped:      " << dstats.deadCount << std::endl;
    out << "  Threaded:     " << dstats.threaded << std::endl;
    out << "  Blocking:     " << dstats.blocking << std::endl;
	out << std::endl;
	}
	if ( m_msgHolder )
	{
	MessageHolder::Stats mstats = getMessageHolderStats();
	out << "Message Holder Stats"  << std::endl;
    out << "  Passes:       " << mstats.passes << std::endl;
    out << "  Holds:        " << mstats.holds << std::endl;
    out << "  Releases:     " << mstats.releases << std::endl;
	out << std::endl;
	}
}

}

}
